package dataflow.core;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.log4j.Logger;

public class ExecutionEventParser implements Runnable {

	static Logger log = Logger.getLogger(ExecutionEventParser.class);
	
	private final LinkedBlockingQueue<ExecutionEvent> queue;
	private AtomicBoolean stop = new AtomicBoolean(false);
	
	private final ExecutorService executor = Executors.newCachedThreadPool();
	
	public ExecutionEventParser(LinkedBlockingQueue<ExecutionEvent> executionQueue) {
		this.queue = executionQueue;
	}
	
	@Override
	public void run() {
		try{
			while(!stop.get()){
				ExecutionEvent e = queue.take();
				e.execute(this, executor);
			}
			executor.shutdown();

		}catch(InterruptedException ie){
			log.debug("Executor interrupted", ie);
		} finally {
			if(!executor.isShutdown()){
				log.debug("ThreadExecutor is going to be shutdown NOW (second attempt).");
				executor.shutdownNow();
			}
			log.debug("I'm leaving, arrivederci!");
		}

	}
	
	public void stop(){
		stop.set(true);
	}


}
